package ex.datastream;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.ArrayList;
import java.util.List;

/**
 * SideOutPut分流，可以多次切分
 */
public class SideOutPut01 extends ApiFrame {
    public static void main(String[] args) throws Exception {
        SideOutPut01 outPut01 = new SideOutPut01();
        outPut01.getEnv();

        //获取数据源
        List data = new ArrayList<Tuple3<Integer, Integer, Integer>>();
        data.add(new Tuple3<>(0, 1, 1));
        data.add(new Tuple3<>(0, 1, 2));
        data.add(new Tuple3<>(0, 1, 3));
        data.add(new Tuple3<>(1, 2, 1));
        data.add(new Tuple3<>(1, 2, 2));
        data.add(new Tuple3<>(1, 4, 3));
        data.add(new Tuple3<>(1, 6, 4));

        DataStream<Tuple3<Integer, Integer, Integer>> items = outPut01.env.fromCollection(data);
        OutputTag<Tuple3<Integer, Integer, Integer>> zeroStream = new OutputTag<Tuple3<Integer, Integer, Integer>>("zeroStream") {
        };
        OutputTag<Tuple3<Integer, Integer, Integer>> oneStream = new OutputTag<Tuple3<Integer, Integer, Integer>>("oneStream") {
        };
        SingleOutputStreamOperator<Tuple3<Integer, Integer, Integer>> processStream = items.process(new ProcessFunction<Tuple3<Integer, Integer, Integer>, Tuple3<Integer, Integer, Integer>>() {
            @Override
            public void processElement(Tuple3<Integer, Integer, Integer> value, Context ctx, Collector<Tuple3<Integer, Integer, Integer>> out) throws Exception {
                if (value.f0 == 0) {
                    ctx.output(zeroStream, value);
                } else if (value.f0 == 1) {
                    ctx.output(oneStream, value);
                }
            }
        });

        DataStream<Tuple3<Integer, Integer, Integer>> zeroSideOutput = processStream.getSideOutput(zeroStream);
        DataStream<Tuple3<Integer, Integer, Integer>> oneSideOutput = processStream.getSideOutput(oneStream);
        zeroSideOutput.printToErr();
        oneSideOutput.print();

        //打印结果
        String jobName="user defined streaming source";
        outPut01.env.execute(jobName);
    }
}
